package cn.snmp.mutual.common;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.PortUnreachableException;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;
import java.util.Set;

import lombok.extern.slf4j.Slf4j;
import org.snmp4j.SNMP4JSettings;
import org.snmp4j.TransportStateReference;
import org.snmp4j.log.LogAdapter;
import org.snmp4j.log.LogFactory;
import org.snmp4j.security.SecurityLevel;
import org.snmp4j.smi.OctetString;
import org.snmp4j.smi.UdpAddress;
import org.snmp4j.transport.DefaultUdpTransportMapping;
import org.snmp4j.util.WorkerTask;

/**
 * @author dengkn
 * XmlFile   XML解析工具
 * @since 2018.9.1
 */
@Slf4j
public class UdpTransportMapping extends DefaultUdpTransportMapping {


    private ListenThread listenerThread;

    private int socketTimeout = 0;

    private int receiveBufferSize = 0;

    private Selector selector = null;
    DatagramChannel channel = null;

    public UdpTransportMapping() throws IOException {
        super(new UdpAddress(InetAddress.getLocalHost(), 0));
        socket = new DatagramSocket(udpAddress.getPort());
        selector = Selector.open();
        channel = DatagramChannel.open();
        channel.configureBlocking(false);
        channel.register(selector, SelectionKey.OP_READ);
    }

    @Override
    public void sendMessage(UdpAddress targetAddress, byte[] message, TransportStateReference tmStateReference)
            throws IOException {
        InetSocketAddress targetSocketAddress = new InetSocketAddress(targetAddress.getInetAddress(),
                targetAddress.getPort());
        if (log.isDebugEnabled()) {
            log.debug("Sending message to " + targetAddress + " with length " + message.length + ": "
                    + new OctetString(message).toHexString());
        }
        DatagramPacket dp = new DatagramPacket(message, message.length, targetSocketAddress);
        ByteBuffer bb = ByteBuffer.wrap(dp.getData(), dp.getOffset(), dp.getLength());
        channel.send(bb, targetSocketAddress);
    }

    public ListenThread getListenerThread() {
        return listenerThread;
    }

    public void setListenerThread(ListenThread listenerThread) {
        this.listenerThread = listenerThread;
    }

    private synchronized DatagramSocket ensureSocket() throws SocketException {
        DatagramSocket s = socket;
        if (s == null) {
            s = new DatagramSocket(udpAddress.getPort());
            s.setSoTimeout(socketTimeout);
            this.socket = s;
        }
        return s;
    }

    @Override
    public synchronized void listen() throws IOException {
        if (listener != null) {
            throw new SocketException("Port already listening");
        }
        ensureSocket();
        listenerThread = new ListenThread();
        listener = SNMP4JSettings.getThreadFactory().createWorkerThread("UdpTransportMapping_" + getAddress(),
                listenerThread, true);
        listener.run();
    }

    class ListenThread implements WorkerTask {

        private byte[] buf;
        private volatile boolean stop = false;

        public ListenThread() throws SocketException {
            buf = new byte[getMaxInboundMessageSize()];
        }

        @Override
        public void run() {
            DatagramSocket socketCopy = socket;
            if (socketCopy != null) {
                try {
                    socketCopy.setSoTimeout(getSocketTimeout());
                    if (receiveBufferSize > 0) {
                        socketCopy.setReceiveBufferSize(Math.max(receiveBufferSize, maxInboundMessageSize));
                    }
                    if (log.isDebugEnabled()) {
                        log.debug("UDP receive buffer size for socket " + getAddress() + " is set to: "
                                + socketCopy.getReceiveBufferSize());
                    }
                } catch (SocketException ex) {
                    log.error("" + ex);
                    setSocketTimeout(0);
                }
            }
            while (!stop) {
                DatagramPacket packet = null;
                ByteBuffer byteBuffer = ByteBuffer.allocate(maxInboundMessageSize);
                try {
                    socketCopy = socket;
                    try {
                        if (socketCopy == null) {
                            stop = true;
                            continue;
                        }
                        if (selector.select() > 0) {
                            Set<SelectionKey> selectedKeys = selector.selectedKeys();
                            Iterator<SelectionKey> iterator = selectedKeys.iterator();
                            while (iterator.hasNext()) {
                                SelectionKey sk = iterator.next();
                                DatagramChannel datagramChannel = (DatagramChannel) sk.channel();
                                iterator.remove();
                                if (sk.isReadable()) {
                                    try {
                                        datagramChannel.receive(byteBuffer);
                                        byteBuffer.flip();
                                        buf = byteBuffer.array();
                                        byteBuffer.clear();
                                    } catch (Exception iox) {
                                        iox.printStackTrace();
                                    }
                                }
                            }
                        }
                        packet = new DatagramPacket(buf, buf.length, udpAddress.getInetAddress(), udpAddress.getPort());
                    } catch (InterruptedIOException iiox) {
                        if (iiox.bytesTransferred <= 0) {
                            continue;
                        }
                    }
                    if (log.isDebugEnabled()) {
                        log.debug("Received message from " + packet.getAddress() + "/" + packet.getPort()
                                + " with length " + packet.getLength() + ": "
                                + new OctetString(packet.getData(), 0, packet.getLength()).toHexString());
                    }
                    ByteBuffer bis;
                    if (isAsyncMsgProcessingSupported()) {
                        byte[] bytes = new byte[packet.getLength()];
                        System.arraycopy(packet.getData(), 0, bytes, 0, bytes.length);
                        bis = ByteBuffer.wrap(bytes);
                    } else {
                        bis = ByteBuffer.wrap(packet.getData());
                    }
                    TransportStateReference stateReference = new TransportStateReference(
                            UdpTransportMapping.this, udpAddress, null, SecurityLevel.undefined,
                            SecurityLevel.undefined, false, socketCopy);
                    fireProcessMessage(new UdpAddress(packet.getAddress(), packet.getPort()), bis, stateReference);
                } catch (SocketTimeoutException stex) {
                } catch (PortUnreachableException purex) {
                    synchronized (UdpTransportMapping.this) {
                        listener = null;
                    }
                    log.error("" + purex);
                    if (log.isDebugEnabled()) {
                        purex.printStackTrace();
                    }
                    if (SNMP4JSettings.isForwardRuntimeExceptions()) {
                        throw new RuntimeException(purex);
                    }
                    break;
                } catch (SocketException soex) {
                    if (!stop) {
                        log.warn("Socket for transport mapping " + toString() + " error: " + soex.getMessage());
                    }
                    if (SNMP4JSettings.isForwardRuntimeExceptions()) {
                        stop = true;
                        throw new RuntimeException(soex);
                    } else {
                        try {
                            DatagramSocket newSocket = renewSocketAfterException(soex, socketCopy);
                            if (newSocket == null) {
                                throw soex;
                            }
                            socket = newSocket;
                        } catch (SocketException e) {
                            stop = true;
                            socket = null;
                            log.error("Socket renewal for transport mapping " + toString() + " failed with: "
                                    + e.getMessage(), e);

                        }
                    }
                } catch (IOException iox) {
                    log.warn("" + iox);
                    if (log.isDebugEnabled()) {
                        iox.printStackTrace();
                    }
                    if (SNMP4JSettings.isForwardRuntimeExceptions()) {
                        throw new RuntimeException(iox);
                    }
                }
            }
            synchronized (UdpTransportMapping.this) {
                listener = null;
                stop = true;
                DatagramSocket closingSocket = socket;
                if ((closingSocket != null) && (!closingSocket.isClosed())) {
                    closingSocket.close();
                }
                socket = null;
            }
            if (log.isDebugEnabled()) {
                log.debug("Worker task stopped:" + getClass().getName());
            }
        }

        public void close() {
            stop = true;
        }

        @Override
        public void terminate() {
            close();
            if (log.isDebugEnabled()) {
                log.debug("Terminated worker task: " + getClass().getName());
            }
        }

        @Override
        public void join() throws InterruptedException {
            if (log.isDebugEnabled()) {
                log.debug("Joining worker task: " + getClass().getName());
            }
        }

        @Override
        public void interrupt() {
            if (log.isDebugEnabled()) {
                log.debug("Interrupting worker task: " + getClass().getName());
            }
            close();
        }
    }
}
